package communication.common.vertx.net;

import communication.common.business.base.FcrSocketAddress;
import communication.common.business.base.HandleMsgData;
import communication.common.business.base.MsgData;
import communication.common.business.hole.HoleConnectRecord;
import communication.common.business.util.MsgDataUtil;
import communication.common.business.util.NetSocketUtil;
import io.vertx.core.*;
import io.vertx.core.buffer.Buffer;
import io.vertx.core.datagram.DatagramPacket;
import io.vertx.core.datagram.DatagramSocket;
import io.vertx.core.json.JsonObject;
import io.vertx.core.net.*;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.io.ObjectOutputStream;
import java.util.Iterator;
import java.util.Map;
import java.util.Objects;
import java.util.Random;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * @author mac
 */
@Slf4j
@Data
public class NetManager {

    private Vertx vertx;

    private HandleMsgData handleMsgData;

    private NetClient netClient;

    private NetServer netServer;

    /**
     * 默认的udp端口连接，用来与server端通信
     */
    private DatagramSocket defaultUdpSocket;

    /**
     * 以远端socket地址为key记录
     */
    private Map<FcrSocketAddress , ConnectRecord<DatagramSocket>> udpRemoteSocketRecord = new ConcurrentHashMap<>();


    /**
     * 打洞记录
     */
    private Map<FcrSocketAddress, HoleConnectRecord> holeRecord = new ConcurrentHashMap<>();

    /**
     * 所有的tcp连接
     */
    private Map<FcrSocketAddress, ConnectRecord<NetSocket>> tcpSockets = new ConcurrentHashMap<>();

    /**
     * 随机数
     */
    public static   Random random = new Random();


    /**
     *  连接的超时时间，默认5秒
     */
    private int timeOut=5000;

    public NetManager(HandleMsgData handleMsgData, Vertx vertx) {
        this.handleMsgData = handleMsgData;
        this.vertx = vertx;
        this.netClient = this.vertx.createNetClient(new NetClientOptions().setConnectTimeout(timeOut));
    }


    /**
     *  connet to address:port
     *
     * @param address
     * @param port
     * @return
     */
    public Future<NetSocket> connectByTcpClient(String address, int port ) throws Exception {
        Promise<NetSocket> promise = Promise.promise();

        if(netClient == null){
            throw new Exception(" netClient not created.");
        }

        netClient.connect(port, address,
                res->{
                    if( res.succeeded()){
                        log.info("tcp-client success connect to  {}:{}",address,port);
                        addTcpSocket(res.result());
                        promise.complete(res.result());
                    }else{
                        log.error("tcp-client fail connect to  {}:{}",address,port);
                    }
                });

        return promise.future();
    }


    public void closeNetSocket(String host, int port,Handler<AsyncResult<Void>> handler){
        FcrSocketAddress remote = new FcrSocketAddress(host, port);
        NetSocket netSocket = getTcpSocketByKey(remote);
        if(netSocket != null ) {
            netSocket.close();
        }
        removeTcpSocket(remote);
    }


    public Future closeUdpServer(FcrSocketAddress remote){
        if(udpRemoteSocketRecord.containsKey(remote)){
            ConnectRecord<DatagramSocket> record = udpRemoteSocketRecord.get(remote);
            udpRemoteSocketRecord.remove(remote);
            return record.getSocket().close();
        }

        return Future.future(null);
    }

    public void sendMsgDataByTcpClient(MsgData msgData , FcrSocketAddress remoteAddress ) throws Exception {
        // 如果连接存在则直接发送消息
        NetSocket socket = getTcpSocketByKey(remoteAddress);
        if( socket != null ){
            NetSocketUtil.sendMsgData(socket,msgData);
            // 连接不存在，那么先建立连接，然后在发送消息
        }else {
            connectByTcpClient(remoteAddress.getHost(), remoteAddress.getPort())
                    .onSuccess(netSocket->{
                        // 记录连接
                        addTcpSocket(netSocket);
                        // 发送消息
                        NetSocketUtil.sendMsgData(netSocket,msgData);
                        // 注册处理tcp的socekt回调内容
                        netSocket.handler(buffer->{
                            handleMsgData.handleAndSendMsg(netSocket);
                        });

                        // 本地socket地址
                        SocketAddress localAddress = netSocket.localAddress();

                        // 启动一个TcpServer监听该端口,以便处理connect请求
                        createTcpServerFuture(localAddress.hostAddress(),localAddress.port());
                    });
        }

    }


    public void sendMsgDataByUdpSocket(MsgData msgData, DatagramSocket socket,FcrSocketAddress remoteAddress,Handler<AsyncResult<Void>>[] sendHandler) throws IOException {
        String msg= Objects.requireNonNull(JsonObject.mapFrom(msgData)).toString();
        Buffer buffer = Buffer.buffer().appendBytes(MsgDataUtil.obj2ByteArray(msgData));
        // 发送消息
        socket.send(buffer, remoteAddress.getPort(), remoteAddress.getHost(), resp -> {
            if (resp.succeeded()) {
                log.debug("udp socket send  msg:{} to address:{} success!", msg, remoteAddress.getAddress());
                // 如果有处理逻辑则进行处理
                if (sendHandler != null && sendHandler[0] != null) {
                    sendHandler[0].handle(resp);
                }

            } else {
                log.debug("send  to :{} msg fail! msg:{},error:{}", remoteAddress.getAddress(), msg, resp.cause().toString());
                if (sendHandler != null && sendHandler[1] != null) {
                    sendHandler[1].handle(resp);
                }
            }
        });

    }

    public void sendMsgDataByUdpServer(MsgData msgData, FcrSocketAddress remoteAddress,Handler<AsyncResult<Void>>[] sendHandler ) throws Exception {
        if( !udpRemoteSocketRecord.containsKey(remoteAddress) ) {
            throw new Exception("udp-socket连接不存在！remoteAddress:"+remoteAddress.getAddress());
        }

        DatagramSocket socket = udpRemoteSocketRecord.get(remoteAddress).getSocket();
        sendMsgDataByUdpSocket(msgData,socket,remoteAddress,sendHandler);
    }


    public  Future createTcpServerFuture( String host, int port){
        return createTcpServer(host,port, handleMsgData::handleAndSendMsg).future();
    }

    public Promise<Boolean> createDefaultUdpServerFuture(FcrSocketAddress localAddress){
        Promise<Boolean> promise = Promise.promise();

        if(defaultUdpSocket !=null ) {
            String msg = "defaultUdpSocket already existed ! ";
            log.warn(msg);
            promise.fail(msg);
            return promise;
        }

        createUdpServerFuture(localAddress).future()
                .onSuccess(socket->{
                    defaultUdpSocket=socket;
                    promise.complete(true);
                })
                .onFailure(promise::fail);
        return promise;
    }


    public Promise<DatagramSocket>  createUdpServerFuture(FcrSocketAddress localAddress){
        Promise<DatagramSocket> promise = Promise.promise();
        DatagramSocket datagramSocket = vertx.createDatagramSocket();
        datagramSocket.listen(localAddress.getPort(), localAddress.getHost(), a -> {
            if (a.succeeded()) {
                log.info("createFcrUdpServer inject handler success! localAddress:{}",localAddress);
                // socket创建完成后放入 udpSocketRecord中进行记录
                a.result().handler(e -> {
                    try {
                        // 处理socket通信内容
                        receivedUdpData(datagramSocket,localAddress, e);
                    } catch (Exception ex) {
                        ex.printStackTrace();
                    }
                });
                promise.complete(datagramSocket);
            } else {
                log.error("createFcrUdpServer inject handler fail:{}", a.cause().getMessage());
                promise.fail(a.cause().getMessage());
            }
        });
        return promise;

    }



    /**
     * 接收到udp的socket数据
     * @param socket
     * @param localAddress
     * @param packet
     * @throws Exception
     */
    private void receivedUdpData(DatagramSocket socket, FcrSocketAddress localAddress, DatagramPacket packet) throws Exception {
        FcrSocketAddress remoteSocketAddress = new FcrSocketAddress(packet.sender().host(),packet.sender().port());
        long now = System.currentTimeMillis();
        // 第一次收到远端的消息
        if( !udpRemoteSocketRecord.containsKey(remoteSocketAddress)) {
            ConnectRecord<DatagramSocket> record = new ConnectRecord<>();
            record.setReceivedFlg(true);
            record.setCreateTime(now);
            record.setRemoteSocketAddress(remoteSocketAddress);
            record.setReceiveTime(now);
            record.setLocalSocketAddress(localAddress);
            record.setProtocolEnum(ProtocolEnum.UDP);
            record.setSocket(socket);
            udpRemoteSocketRecord.put(remoteSocketAddress,record);
        }else {
            // 不是第一次，更新消息的接收时间即可
            ConnectRecord<DatagramSocket> record = udpRemoteSocketRecord.get(remoteSocketAddress);
            record.setReceiveTime(now);
        }

        // 检查是否是打洞用的端口接收到的数据，如果是打洞，则设置打洞记录的标识为true
         if (holeRecord.containsKey(remoteSocketAddress)){
             HoleConnectRecord connectRecord = holeRecord.get(remoteSocketAddress);
             if(connectRecord.getHoleFinish().get()){
                 // 该打洞记录已经处理成功，则跳过处理
                 return;
             }
             AtomicBoolean holeFinish = connectRecord.getHoleFinish();
             connectRecord.setLocalPortOnSuccess(localAddress.getPort());
             holeFinish.set(true);

         }

        // 处理socket的通信内容
        handleMsgData.handleAndSendMsg(socket, packet);
    }
    
    

    /**
     * 创建udp的connect记录
     * @param socket socket连接
     * @param localAddress  本地地址
     * @param remoteAddress 远端地址
     * @return
     */
    private ConnectRecord<DatagramSocket> createUdpConnectRecord(DatagramSocket socket , FcrSocketAddress localAddress, FcrSocketAddress remoteAddress) {
        ConnectRecord<DatagramSocket> record = new ConnectRecord<>();
        record.setCreateTime(System.currentTimeMillis());
        record.setProtocolEnum(ProtocolEnum.UDP);
        record.setLocalSocketAddress(localAddress);
        record.setRemoteSocketAddress(remoteAddress);
        record.setSocket(socket);
        return record;
    }


    public DatagramSocket getUdpSocketByKey(FcrSocketAddress key){
        return udpRemoteSocketRecord.get(key).getSocket();
    }

    public NetSocket getTcpSocketByKey(FcrSocketAddress key){
        return tcpSockets.get(key).getSocket();
    }


    public void addTcpSocket(NetSocket socket){
        FcrSocketAddress key = NetSocketUtil.getSocketAddress(socket.remoteAddress());
        log.info("tcp socket Address:{}",key);
        addTcpSocket(key,socket);
    }


    // todo:
    private void addTcpSocket(FcrSocketAddress key, NetSocket socket) {
        if(tcpSockets.containsKey(key)){
            log.warn("tcp socket address already existed! key:{}",key);
        }
//        tcpSockets.put(key, socket);
    }


    public void removeTcpSocket(FcrSocketAddress key){
        if(!tcpSockets.containsKey(key)){
            log.warn("can not remove udp socket,invalid key:{}",key);
            return;
        }
        tcpSockets.remove(key);
    }

    public Iterator<Map.Entry<FcrSocketAddress,ConnectRecord<NetSocket>>> getTcpSocketIterator(){
        return tcpSockets.entrySet().iterator();
    }


    public Iterator<Map.Entry<FcrSocketAddress,ConnectRecord<DatagramSocket>>> getUdpSocketIterator(){
        return udpRemoteSocketRecord.entrySet().iterator();
    }

    /**
     * 创建tcpServer
     * @param port
     * @return
     */
    public Promise<Boolean> createTcpServer(String host, int port, Handler<NetSocket> socketHandler){
        Promise<Boolean> promise = Promise.promise();
        netServer = vertx.createNetServer();
        // 设置处理处理连接的handler
        netServer.connectHandler(netSocket->{
            // 连接成功就保存连接记录
            addTcpSocket(netSocket);
            socketHandler.handle(netSocket);
        }  );
        // 监听指定的端口和host
        netServer.listen(port, host, result->{
            if(result.succeeded()){
                log.info("FcrTcpServer listen host:{} ,port:{}  success!",host,port);
                promise.complete(true);
            }else {
                log.error("FcrTcpServer listen port:{} fail !",port);
                netServer.close();
                promise.complete(false);
            }
        });

        return promise;
    }


    public void stopTcpServer(Handler<AsyncResult<Void>> completionHandler){
        if(netServer == null){
            return;
        }
        netServer.close(completionHandler);
    }

    public void holePunchingOnUdp( FcrSocketAddress remoteAddress ,int startPort,String localHost,int count) throws Exception {

        MsgData msgData = MsgDataUtil.createHoleReq( random.nextInt() );

        HoleConnectRecord holeConnectRecord = initHoleConnectRecord(remoteAddress);

        holeRecord.put(remoteAddress,holeConnectRecord);

        // 建立1000个监听端口,用这1000个端口给远端发送心跳消息
        for (int i=0 ; i<count; i++ ){
            if(holeConnectRecord.getHoleFinish().get()){
                log.info("[holeFinish为true,跳过剩余打洞流程] remoteAddress:{} ",remoteAddress);
                return;
            }
            int port = startPort+i;
            FcrSocketAddress localAddress = new FcrSocketAddress(localHost,port);
            createUdpServerFuture(localAddress)
                    .future()
                    .onSuccess(socket->{
                        // 监听端口成功，记录打洞数据和socket
                        holeConnectRecord.getLocalPortMap().put(port,socket);

                        // 新建立的socket发送消息, 这里的二次是检查是为了处理在循环中下一个端口刚创建完成时，可能之前的某个端口打洞成功的场景
                        if(!holeConnectRecord.getHoleFinish().get()){
                            try {
                                NetSocketUtil.sendObj(socket,remoteAddress.getHost(),remoteAddress.getPort(),msgData);
                            } catch (IOException e) {
                                e.printStackTrace();
                            }
                        }
                    });
        }
    }

    private HoleConnectRecord initHoleConnectRecord(FcrSocketAddress remoteAddress) {
        HoleConnectRecord  holeConnectRecord = new HoleConnectRecord();
        holeConnectRecord.setCreateTime(System.currentTimeMillis());
        holeConnectRecord.setRemoteSocketAddress(remoteAddress);
        holeConnectRecord.setTimeOut(timeOut);
        return holeConnectRecord;
    }

    /**
     * 处理超时的打洞记录数据
     */
    public void  holeTimeOutHoleConnectRecord(){
        while (true){
            Iterator<HoleConnectRecord> iterator = holeRecord.values().iterator();
            if(iterator.hasNext()){

                HoleConnectRecord next = iterator.next();
                long now = System.currentTimeMillis();
                // 超时
                if(now >= next.getTimeOut()+next.getCreateTime()){
                    // 移除元素
                    iterator.remove();
                    // 关闭socket
                    if(!next.getHoleFinish().get()){
                        // 打洞未成功，所有socket全部close
                        next.getLocalPortMap().entrySet().forEach(e->{
                            e.getValue().close();
                            log.info("[打洞处理]-未收到对端通信，关闭本地监听端口:{}",e.getKey());
                        });

                    }else {
                        //打洞成功，关闭其他连接失败的socket
                        next.getLocalPortMap().entrySet()
                                .stream()
                                .filter(e-> !e.getKey().equals(next.getLocalPortOnSuccess()))
                                .map(Map.Entry::getValue)
                                .forEach(DatagramSocket::close);
                    }

                }
            }

        }
    }

}
